package cn.xuexiyuan.flinkstudy.transformation;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Description: 拆分和选择操作
 *
 * @Author 左龙龙
 * @Date 21-3-23
 * @Version 1.0
 **/
public class TransformationDemo03 {
    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        // env.generateSequence(开始, 结束)
        DataStreamSource<Long> ds = env.fromSequence(0, 10);

        // 2.transformation
        // 对流中的数据进行奇偶数区分
        OutputTag<Long> oddTag = new OutputTag<>("奇数", TypeInformation.of(Long.class));
        OutputTag<Long> evenTag = new OutputTag<>("偶数", TypeInformation.of(Long.class));

        SingleOutputStreamOperator<Long> process = ds.process(new ProcessFunction<Long, Long>() {
            @Override
            public void processElement(Long aLong, Context context, Collector<Long> collector) throws Exception {
                if (aLong % 2 == 0) {
                    context.output(evenTag, aLong);
                } else {
                    context.output(oddTag, aLong);
                }
            }
        });

        DataStream<Long> oddSideOutput = process.getSideOutput(oddTag);
        DataStream<Long> evenSideOutput = process.getSideOutput(evenTag);

        // 3.sink
        oddSideOutput.print("【奇数】");
        evenSideOutput.print("【偶数】");

        // 4.excute
        env.execute("TransformationDemo03");

    }
}
